package dr.server.consumer;

import com.alibaba.fastjson.JSONObject;
import dr.common.vo.PusherMessage;
import dr.server.impl.PusherService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class PusherConsumer {

    @Value("${spring.profiles.active}")
    String active = "dev";

    @Autowired
    PusherService pusherService;

    /**
     * 解析消息
     * type： 消息类型（0=不发送 1=阿里云短信 2=赛迪短信 3=极光推送)
     * msg： 发送内容
     * @param records 消息队列信息
     */
    @KafkaListener(id = "pusher-0", topics = "ZSKJ-pusher", groupId="ZSKJ-pusher")
    public void listen(ConsumerRecord<?, ?> records) {
        log.debug("===================================================接收到消費信息：" + records.toString());
        try {
            if ("prod".equals(active)||"test".equals(active)) {
                if(null != records.value()) {
                    PusherMessage message = JSONObject.parseObject(records.value().toString(), PusherMessage.class);
                    pusherService.doPush(message);
                }
            }else {
                log.error("非生产环境默认发送成功:{}",records.toString());
            }
        }catch (Exception e) {
            log.error("消费失败:{},{}",records.toString(), e.getLocalizedMessage());
        }

    }
}